package org.sabayframework.activity.impl;

import org.sabayframework.activity.ActivityContext;
import org.sabayframework.channel.Channel;
import org.sabayframework.channel.Lock;

/**
 * Blocking activity blocks both input and output channels for time of
 * execution.
 * */
public abstract class BlockingActivity<I, O> extends AbstractActivity<I, O> {

	@Override
	public final void execute(ActivityContext<I, O> ctx) throws Exception {
		final Channel<I> input = ctx.getInput();
		final Channel<O> output = ctx.getOutput();
		int available = 0;
		do
		{
			available = input.size();
			if(available == 0) return;
			final int freeSize = output.freeSize();
			if(available > freeSize) available = freeSize;
			if(available > 0){
				Lock<I> iLock=input.getReadLock(available);
				if(iLock!=null){
					Lock<O> oLock = output.getWriteLock(available);
					if(oLock==null) iLock.cancel();
					else {
						try{
							execute(ctx, iLock, oLock);
						}finally{
							iLock.unlock();
							oLock.unlock();
						}
						return;
					}
				}
			}
		} while(available > 0);
	}
	
	/**
	 * Implementor should define activity body.
	 * @param iLock input lock. Sizes of both locks are equal.
	 * @param oLock output lock. Sizes of both locks are equal.
	 * */
	public abstract void execute(ActivityContext<I, O> ctx, Lock<I> iLock, Lock<O> oLock) throws Exception;

}
